package com.aotain.project.apollo.bolt;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 
*  @ClassName: SourceFmtBolt 
* @Description: TODO(这里用一句话描述这个类的作用) 
* @author 程彬
* @date 2015年8月7日 下午2:56:10 
*
 */
public class SourceFmtBolt_backup implements IRichBolt {

	OutputCollector collector = null;
	private Set<String> IPSet = null;//要过滤的ip
	
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","dport","sip","sport"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

//	@Override
//	public void prepare(Map stormConf, TopologyContext context) {
//		// TODO Auto-generated method stub
//
//		System.err.println("prepare");
//	}

//	@Override
//	public void execute(Tuple tuple, BasicOutputCollector collector) {
//		// TODO Auto-generated method stub
//		try {
////			System.err.println("begin sourcefmtbolt");
//			String logString = tuple.getString(0);
//			if(logString != null && !"".equals(logString)) {
//				String[] splits = logString.split("\\|",-1);
//				if(splits.length>9) {
//					String dip = splits[2];
//					String dport = splits[5];
//					String sip = splits[1];
//					String sport = splits[4];
//					if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport)) {
////						System.err.println("log" + dip + " " + dport + " " + sip );
//						collector.emit(new Values(dip,dport,sip,sport));
//					}
//				}
//			} else {
//				System.err.println("LogLine is NULL……");
//			}
//		} catch (Exception e) {
//			e.printStackTrace();
//			throw new FailedException("SourceFmtBolt消息处理失败！");
//		}
//	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
		String ips = "101.227.130.26,180.163.28.22,101.227.160.27,61.151.186.3,182.254.150.24,182.254.139.240,203.195.193.178,182.254.219.60,182.254.225.173,61.188.37.186,182.254.241.239,115.159.30.208,222.73.33.114,222.73.33.115,222.73.33.116,101.226.182.53,101.226.182.92,222.73.155.185,101.226.182.75,222.73.195.213,183.60.170.217,101.226.64.227,115.159.26.254,101.226.131.205,101.226.69.253,183.62.140.141,113.116.229.21";
		String[] ipsArray = ips.split(",");
		IPSet = new HashSet<String>(Arrays.asList(ipsArray));//数组转为set
		
	}

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		String logString = tuple.getString(0);
		if(logString != null && !"".equals(logString)) {
			String[] splits = logString.split("\\|",-1);
			if(splits.length>9) {
				String dip = splits[2];
				String dport = splits[5];
				String sip = splits[1];
				String sport = splits[4];
				if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport) && IPSet.contains(dip)) {
//					System.err.println("log" + dip + " " + dport + " " + sip );
					collector.emit(new Values(dip,dport,sip,sport));
					collector.ack(tuple);
				}
			}
		}
	}

}
